package com.atguigu.bigdata.spark.core.rdd.operator.transform;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.sources.In;
import scala.Tuple2;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Iterator;
import java.util.List;

// 从服务器日志数据apache.log中获取每个时间段访问
public class Spark06_RDD_Operator_Transform_Test_JAVA {
    public static void main(String[] args) {

        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("sparkCore");
        JavaSparkContext sc = new JavaSparkContext(conf);

        JavaRDD<String> rdd = sc.textFile("datas/apache.log");
        JavaRDD<String> map = rdd.map(new Function<String, String>() {
            @Override
            public String call(String s) throws Exception {
                String[] temp = s.split(" ");
                String input = temp[3];
                SimpleDateFormat sdf = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss");
                Date date = sdf.parse(input);
                SimpleDateFormat sdf1 = new SimpleDateFormat("HH");
                String hour = sdf1.format(date);
                return hour;
            }
        });

        JavaPairRDD<String, Iterable<String>> group = map.groupBy(new Function<String, String>() {
            @Override
            public String call(String v1) throws Exception {
                return v1;
            }
        });

        JavaRDD<Tuple2<String, Integer>> mapResRdd = group.map(new Function<Tuple2<String, Iterable<String>>, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> call(Tuple2<String, Iterable<String>> v1) throws Exception {
                Iterator<String> iterator = v1._2.iterator();
                int res = 0;
                while(iterator.hasNext()){
                    iterator.next();
                    res++;
                }
                return new Tuple2<String, Integer>(v1._1(), res);
            }
        });

        List<Tuple2<String, Integer>> res = mapResRdd.collect();

        for(Tuple2<String, Integer> val : res) {
            System.out.println(val);
        }
        sc.stop();
    }
}
